package com.nightsoul.hadoop1.test.counter;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobID;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.Task;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.OldJobBuilder;

public class MissingTemperatureFields extends Configured implements Tool {

	@Override
	public int run(String[] args) throws Exception {
		if(args.length!=1) {
			OldJobBuilder.printUsage(this, "<job ID>");
			return -1;
		}
		
		JobClient client = new JobClient(new JobConf(getConf()));
		String jobId = args[0];
		RunningJob job = client.getJob(JobID.forName(jobId));
		if(job==null) {
			System.err.printf("No job with ID %s found.\n", jobId);
			return -1;
		}
		
		if(!job.isComplete()) {
			System.err.printf("Job %s is not complete.\n", jobId);
			return -1;
		}
		
		Counters counters = job.getCounters();
		long missing = counters.getCounter(MaxTemperatureWithCounter.Temperature.MISSING);
		long total = counters.findCounter(Task.Counter.MAP_INPUT_RECORDS).getCounter();
		System.out.printf("Records with missing temperature fields: %.2f%%\n ", 100.0*missing/total);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new MissingTemperatureFields(), args);
		System.exit(exitCode);
	}
}
